package com.zuipin.service.impl;

import java.util.List;

import javax.annotation.Resource;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerOrderly;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
import com.zuipin.entity.RocketMqErrorInfo;
import com.zuipin.entity.RocketMqInfo;
import com.zuipin.enums.RocketMQTagEnum;
import com.zuipin.mapper.RocketMqErrorInfoMapper;
import com.zuipin.mapper.RocketMqInfoMapper;
import com.zuipin.service.IProductInfoService;
import com.zuipin.service.IRocketMQService;
import com.zuipin.util.CommonUtils;
import com.zuipin.util.DateUtils;
import com.zuipin.util.StringUtil;

/**
 * @ClassName: RocketMQService
 * @author xujinding
 * @time 2016年11月16日下午7:46:08
 */
@Service
public class RocketMQServiceImpl implements IRocketMQService {
	
	private final static Logger		log			= Logger.getLogger(RocketMQServiceImpl.class);
	@Resource
	private RocketMqInfoMapper		rocketMqInfoMapper;
	@Resource
	private RocketMqErrorInfoMapper	rocketMqErrorInfoMapper;
	
	@Resource
	private IProductInfoService		productInfoService;
	
	@Value("${rocketmq.namesrvAddr}")
	public String					NAMESRVADDR	= null;											// 服务器地址
	
	@Value("${rocketmq.userName}")
	public String					USERNAME	= null;											// 使用者
	private static byte[]			lock		= new byte[0];									// 数组对象的锁
	
	/**
	 * 保存错误信息
	 * 
	 * @Title: saveError
	 * @author: xujinding
	 * @date: 2017年6月8日下午4:01:16
	 * @param id
	 * @param message
	 * @return: void
	 * @throws
	 */
	public RocketMqErrorInfo saveError(Long infoId, String errorInfo) {
		if (StringUtil.isNotBlank(errorInfo)) {
			if (errorInfo.length() > 1000) {
				errorInfo = errorInfo.substring(0, 990);
			}
		} else {
			errorInfo = "错误信息为空";
		}
		log.error(errorInfo);
		RocketMqErrorInfo rocketMqErrorInfo = new RocketMqErrorInfo();
		rocketMqErrorInfo.setNum(1);
		rocketMqErrorInfo.setMqInfoId(infoId == null ? null : infoId);
		rocketMqErrorInfo.setErrorInfo(errorInfo);
		rocketMqErrorInfo.setState(RocketMqErrorInfo.CREATE_NEW);
		rocketMqErrorInfo.setCreatedTime(DateUtils.getCurrentDateTime());
		rocketMqErrorInfo.setUpdateTime(DateUtils.getCurrentDateTime());
		rocketMqErrorInfoMapper.insertSelective(rocketMqErrorInfo);
		return rocketMqErrorInfo;
	}
	
	/**
	 * 保存发送信息
	 * 
	 * @Title: saveError
	 * @author: xujinding
	 * @date: 2017年6月8日下午4:01:16
	 * @param id
	 * @param message
	 * @return: void
	 * @throws
	 */
	/* (non-Javadoc)
	 * @see com.zuipin.service.IRocketMQService#saveRocketMQInfo(java.lang.String, java.lang.String, java.lang.String, java.lang.String, java.lang.Integer, java.lang.String, java.lang.Integer, java.lang.String, java.lang.String)
	 */
	public RocketMqInfo saveRocketMQInfo(String topic, String tag, String key, String body, Integer type, String msgId, Integer queueId, String bornHost, String storeHost) {
		
		RocketMqInfo temp = new RocketMqInfo();
		temp.setType(type);
		temp.setKeyNo(key);
		RocketMqInfo rocketMQInfo = rocketMqInfoMapper.findBySelect(temp);
		// 判断当前消息是否存在
		if (rocketMQInfo == null) {
			rocketMQInfo = new RocketMqInfo();
			rocketMQInfo.setIsExist(false);
		} else {
			// 如果消息已存在并已经被消费过了，那么消息不在保存
			rocketMQInfo.setIsExist(true);
			return rocketMQInfo;
		}
		
		rocketMQInfo.setNums(1);
		rocketMQInfo.setTopic(topic);
		rocketMQInfo.setTag(tag);
		rocketMQInfo.setKeyNo(key);
		rocketMQInfo.setBody(body);
		rocketMQInfo.setType(type);
		rocketMQInfo.setMsgId(msgId);
		rocketMQInfo.setQueueId(queueId);
		rocketMQInfo.setBornHost(bornHost);
		rocketMQInfo.setStoreHost(storeHost);
		rocketMqInfoMapper.insertSelective(rocketMQInfo);
		return rocketMQInfo;
	}
	
	/**
	 * 同步发送
	 * 
	 * @Title: doSendSyncMsg
	 * @author: wuyicheng
	 * @date: 2016年11月18日下午2:25:49
	 * @Description: TODO
	 * @param producerGroup
	 *            生产组
	 * @param instanceName
	 *            实例名
	 * @param topicName
	 *            主题
	 * @param tagName
	 *            次标题
	 * @param key
	 *            唯一标识
	 * @param json
	 *            发送的消息内容
	 * @return String 成功标志 SEND_OK
	 * @throws
	 */
	public String doSendSyncMsg(String producerGroup, String instanceName, String topicName, String tagName, String key, String json) {
		topicName = topicName + USERNAME;
		RocketMqInfo rocketMqInfo = saveRocketMQInfo(topicName, tagName, key, json, //
				RocketMqInfo.TYPE_PRODUCER_1, null, null, null, null);
		DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
		SendResult sendResult = new SendResult();
		String str = "";
		try {
			// 生产者放置InstanceName 有利于多线程发送
			producer.setInstanceName(CommonUtils.getUUID());
			producer.setNamesrvAddr(NAMESRVADDR);
			// 发送端是否重发
			producer.setRetryAnotherBrokerWhenNotStoreOK(true);
			producer.start();
			Message msg = new Message(topicName, tagName, key, json.getBytes());
			// 消息延迟
			// msg.setDelayTimeLevel(10);
			// 发送消息
			sendResult = producer.send(msg);
			log.info("发送结果:" + sendResult);
			
		} catch (Exception e) {
			e.printStackTrace();
			saveError(rocketMqInfo.getId(), e.getMessage());
		} finally {
			producer.shutdown();
		}
		if (sendResult != null) {
			str = sendResult.getSendStatus().toString();
		}
		return str;
	}
	
	/**
	 * 消息重新发送
	 * 
	 * @Title: doReSendMsg
	 * @author: xujinding
	 * @date: 2017年6月8日下午2:21:28
	 * @Description:
	 * @return: void
	 * @throws
	 */
	public void doReSendMsg() {
		List<RocketMqInfo> list = rocketMqInfoMapper.findReSendMsg();
		for (RocketMqInfo rocketMQInfo : list) {
			// 重发后修改重发状态
			rocketMQInfo.setIsResend(RocketMqInfo.IS_RESEND_NO_0);
			rocketMQInfo.setUpdateTime(DateUtils.getCurrentDateTime());
			rocketMqInfoMapper.updateByPrimaryKeySelective(rocketMQInfo);
			if (rocketMQInfo.getTopic() == null) {
				continue;
			}
			
			String group = RocketMqInfo.SYSTEM_NAME + RocketMqInfo.PRODUCT_GROUP;
			String name = RocketMqInfo.SYSTEM_NAME + RocketMqInfo.PRODUCT_NAME;
			String topic = rocketMQInfo.getTopic();
			if (topic.indexOf("_") != -1) {
				topic = topic.substring(0, topic.indexOf("_"));
			}
			String tag = rocketMQInfo.getTag();
			String key = "R-" + CommonUtils.getUUID();// 重发消息
			String body = rocketMQInfo.getBody();
			doSendSyncMsg(group, name, topic, tag, key, body);
		}
	}
	
	/**
	 * 顺序消费
	 * 
	 * @Title: doOrderlyConsumer
	 * @author: xujinding
	 * @date: 2017年6月9日下午2:23:02
	 * @param consumerGroup
	 *            消费组
	 * @param instanceName
	 *            消费者实例名
	 * @param topicName
	 *            主题
	 * @param tagName
	 *            次主题(多个标题用逗号隔开)
	 * @param model
	 *            消费方式：MessageModel.CLUSTERING 集群 MessageModel.BROADCASTING 广播
	 * @throws
	 */
	public void doOrderlyConsumer(String consumerGroup, String instanceName, String topicName, String tagName, MessageModel model) {
		consumerGroup = consumerGroup + USERNAME;
		topicName = topicName + USERNAME;
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
		consumer.setNamesrvAddr(NAMESRVADDR);
		consumer.setMessageModel(MessageModel.CLUSTERING);
		// 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br> 如果非第一次启动，那么按照上次消费的位置继续消费
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
		if (StringUtil.isNotBlank(tagName)) {
			tagName = tagName.replace(",", "||");
		} else {
			tagName = "*";
		}
		// 订阅指定topic下所有消息<br> 注意：一个consumer对象可以订阅多个topic
		try {
			consumer.subscribe(topicName, tagName);
			log.info("@@@@@@@@@@@@@@@@@当前订阅的是" + topicName + "@@@@@@@@@@@@@@@@@@@@@@@@");
		} catch (MQClientException e) {
			log.error(e.getMessage());
		}
		
		consumer.registerMessageListener(new MessageListenerOrderly() {
			public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
				
				if (msgs != null && msgs.size() > 0) {
					for (MessageExt messageExt : msgs) {
						/**
						 * 添加业务代码
						 */
						log.info("@@@@@@@@@@@@@@@@@开始执行业务@@@@@@@@@@@@@@@@@@@@@@@");
						boolean flag = doExecute(messageExt);
						if (flag) {
							return ConsumeOrderlyStatus.SUCCESS;
						}
						return ConsumeOrderlyStatus.SUCCESS;
					}
				}
				return ConsumeOrderlyStatus.SUCCESS;
			}
			
		});
		try {
			consumer.start();
		} catch (MQClientException e) {
			log.error(e.getMessage());
		}
		log.info("-----------组:" + consumerGroup + "开启默认实例--------------------------");
	}
	
	/**
	 * @Description:根据二级主题执行不同业务流程
	 * @Title: doExecute
	 * @author: wuyicheng
	 * @date: 2016年11月23日上午9:21:26
	 * @param messageExt
	 * @return
	 * @return: boolean 执行结果
	 * @throws
	 */
	private boolean doExecute(MessageExt messageExt) {
		RocketMqInfo rocketMQInfo = null;
		Long id = null;
		try {
			synchronized (lock) {
				rocketMQInfo = saveRocketMQInfo(messageExt.getTopic(), messageExt.getTags(), messageExt.getKeys(), new String(messageExt.getBody()), RocketMqInfo.TYPE_CONSUMER_2,
						messageExt.getMsgId(), messageExt.getQueueId(), messageExt.getBornHost().toString(), messageExt.getStoreHost().toString());
			}
			// 根据keyno查询消息如果存在就不往下执行
			if (rocketMQInfo != null && rocketMQInfo.getIsExist() != null && rocketMQInfo.getIsExist()) {
				log.error(messageExt.getKeys() + ",消息重复！");
				return true;
			}
			id = rocketMQInfo.getId();
			if (messageExt.getTags().equals(RocketMQTagEnum.TO_UPDATE_ERP_ID.toString())) {
				productInfoService.updateProductErpId(rocketMQInfo);
			} else {
				log.info("OMS未订阅" + messageExt.getTags());
			}
			// 业务处理成功后状态更新为不重发
			updateIsReSend(id, RocketMqInfo.IS_RESEND_NO_0, RocketMqInfo.TYPE_CONSUMER_2);
		} catch (Exception e) {
			log.error("消息消费失败--保存重发状态开始", e);
			// 执行失败 且不是重发的消息 自动保存为重发状态
			if (rocketMQInfo != null && rocketMQInfo.getKeyNo() != null && rocketMQInfo.getKeyNo().indexOf(RocketMqInfo.RESEND_PRE_FIX) < 0) {
				updateIsReSend(id, RocketMqInfo.IS_RESEND_YES_1, RocketMqInfo.TYPE_CONSUMER_2);
			} else {
				updateIsReSend(id, RocketMqInfo.IS_RESEND_NO_0, RocketMqInfo.TYPE_CONSUMER_2);
			}
			saveError(id, e.getMessage());
			
		}
		return true;
	}
	
	/**
	 * 更新是否重发
	 * 
	 * @Title: updateIsReSend
	 * @author joy.zhang
	 * @date 2017年1月20日 上午9:17:06
	 * @param id
	 * @param isReSend
	 * @param typeConsumer
	 * @return void
	 */
	public void updateIsReSend(Long id, Integer isReSend, Integer typeConsumer) {
		RocketMqInfo info = rocketMqInfoMapper.selectByPrimaryKey(id);
		info.setIsResend(isReSend);
		info.setType(typeConsumer);
		rocketMqInfoMapper.updateByPrimaryKeySelective(info);
	}
}
